package com.flying.fish.gateway.filter.factory;

import com.netflix.hystrix.HystrixCommandGroupKey;
import com.netflix.hystrix.HystrixCommandKey;
import com.netflix.hystrix.HystrixObservableCommand;
import com.netflix.hystrix.HystrixCommandGroupKey.Factory;
import com.netflix.hystrix.HystrixObservableCommand.Setter;
import com.netflix.hystrix.exception.HystrixRuntimeException;
import com.netflix.hystrix.exception.HystrixRuntimeException.FailureType;
import java.net.URI;
import java.util.Arrays;
import java.util.List;
import java.util.function.Consumer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.cloud.gateway.filter.GatewayFilter;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.factory.AbstractGatewayFilterFactory;
import org.springframework.cloud.gateway.support.ServerWebExchangeUtils;
import org.springframework.cloud.gateway.support.TimeoutException;
import org.springframework.core.annotation.AnnotatedElementUtils;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.reactive.DispatcherHandler;
import org.springframework.web.server.ResponseStatusException;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.util.UriComponentsBuilder;
import reactor.core.publisher.Mono;
import rx.Observable;
import rx.RxReactiveStreams;
import rx.Subscription;
import rx.functions.Action1;

@Slf4j
@Component
public class HystrixGatewayFilterFactory extends AbstractGatewayFilterFactory<HystrixGatewayFilterFactory.Config> {
    public static final String FALLBACK_URI = "fallbackUri";
    private final ObjectProvider<DispatcherHandler> dispatcherHandlerProvider;
    private volatile DispatcherHandler dispatcherHandler;

    public HystrixGatewayFilterFactory(ObjectProvider<DispatcherHandler> dispatcherHandlerProvider) {
        super(HystrixGatewayFilterFactory.Config.class);
        this.dispatcherHandlerProvider = dispatcherHandlerProvider;
    }

    private DispatcherHandler getDispatcherHandler() {
        if (this.dispatcherHandler == null) {
            this.dispatcherHandler = (DispatcherHandler)this.dispatcherHandlerProvider.getIfAvailable();
        }

        return this.dispatcherHandler;
    }

    public List<String> shortcutFieldOrder() {
        return Arrays.asList("name");
    }

    public GatewayFilter apply(String routeId, Consumer<HystrixGatewayFilterFactory.Config> consumer) {
        HystrixGatewayFilterFactory.Config config = (HystrixGatewayFilterFactory.Config)this.newConfig();
        consumer.accept(config);
        if (StringUtils.isEmpty(config.getName()) && !StringUtils.isEmpty(routeId)) {
            config.setName(routeId);
        }

        return this.apply(config);
    }

    public GatewayFilter apply(HystrixGatewayFilterFactory.Config config) {
        if (config.setter == null) {
            Assert.notNull(config.name, "A name must be supplied for the Hystrix Command Key");
            HystrixCommandGroupKey groupKey = Factory.asKey(this.getClass().getSimpleName());
            HystrixCommandKey commandKey = com.netflix.hystrix.HystrixCommandKey.Factory.asKey(config.name);
            config.setter = Setter.withGroupKey(groupKey).andCommandKey(commandKey);
        }

        return (exchange, chain) -> {
            HystrixGatewayFilterFactory.RouteHystrixCommand command = new HystrixGatewayFilterFactory.RouteHystrixCommand(config.setter, config.fallbackUri, exchange, chain);
            return Mono.create((s) -> {
                Subscription sub = command.toObservable().subscribe(s::success,
                        s::error, s::success);
                // 3.Mono取消时，取消Observable<HystrixCommand>的订阅，结束HystrixCommand的执行
                s.onCancel(sub::unsubscribe);
            }).onErrorResume((throwable) -> {
                if (throwable instanceof HystrixRuntimeException) {
                    HystrixRuntimeException e = (HystrixRuntimeException)throwable;
                    FailureType failureType = e.getFailureType();
                    switch(failureType) {
                        case TIMEOUT:
                            return Mono.error(new TimeoutException());
                        case COMMAND_EXCEPTION:
                            Throwable cause = e.getCause();
                            if (cause instanceof ResponseStatusException || AnnotatedElementUtils.findMergedAnnotation(cause.getClass(), ResponseStatus.class) != null) {
                                return Mono.error(cause);
                            }
                    }
                }

                return Mono.error(throwable);
            }).then();
        };
    }

    public static class Config {
        private String name;
        private Setter setter;
        private URI fallbackUri;

        public Config() {
        }

        public String getName() {
            return this.name;
        }

        public HystrixGatewayFilterFactory.Config setName(String name) {
            this.name = name;
            return this;
        }

        public HystrixGatewayFilterFactory.Config setFallbackUri(String fallbackUri) {
            if (fallbackUri != null) {
                this.setFallbackUri(URI.create(fallbackUri));
            }

            return this;
        }

        public URI getFallbackUri() {
            return this.fallbackUri;
        }

        public void setFallbackUri(URI fallbackUri) {
            if (fallbackUri != null && !"forward".equals(fallbackUri.getScheme())) {
                throw new IllegalArgumentException("Hystrix Filter currently only supports 'forward' URIs, found " + fallbackUri);
            } else {
                this.fallbackUri = fallbackUri;
            }
        }

        public HystrixGatewayFilterFactory.Config setSetter(Setter setter) {
            this.setter = setter;
            return this;
        }
    }

    private class RouteHystrixCommand extends HystrixObservableCommand<Void> {
        private final URI fallbackUri;
        private final ServerWebExchange exchange;
        private final GatewayFilterChain chain;

        RouteHystrixCommand(Setter setter, URI fallbackUri, ServerWebExchange exchange, GatewayFilterChain chain) {
            super(setter);
            this.fallbackUri = fallbackUri;
            this.exchange = exchange;
            this.chain = chain;
        }

        protected Observable<Void> construct() {
            return RxReactiveStreams.toObservable(this.chain.filter(this.exchange));
        }

        protected Observable<Void> resumeWithFallback() {
            if (this.fallbackUri == null) {
                return super.resumeWithFallback();
            } else {
                URI uri = this.exchange.getRequest().getURI();
                boolean encoded = ServerWebExchangeUtils.containsEncodedParts(uri);
                URI requestUrl = UriComponentsBuilder.fromUri(uri).host((String)null).port((String)null).uri(this.fallbackUri).scheme((String)null).build(encoded).toUri();
                this.exchange.getAttributes().put(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR, requestUrl);
                ServerHttpRequest request = this.exchange.getRequest().mutate().uri(requestUrl).build();
                ServerWebExchange mutated = this.exchange.mutate().request(request).build();
                return RxReactiveStreams.toObservable(HystrixGatewayFilterFactory.this.getDispatcherHandler().handle(mutated));
            }
        }
    }
}

